补注喵

For Proj-04

并发编程

Python 中的并发有三种形式，多线程、多进程和多协程^[1]。其中多线程和多协程适合 IO 密集型的场合，多进程则更适合 CPU 密集型的场合。

在对 IO 要求极高的场合，可以将它们结合起来使用^[2]。由于 GIL 的存在，Python 中的 CPU 密集型任务一般会通过多进程完成^[3]。

可阅读这篇博客了解更多信息 w

这次我们使用多线程来应对并发网络请求。

多线程

Python 中一般使用标准库 threading 进行多线程操作。参见官方文档。

线程能够共享相同的内存和资源，所以需要处理访问冲突（竞争条件 / 数据竞争）的问题。在多个线程可能同时读写共享资源时需要加锁。

很多时候直接把可调用对象作为 Thread 类的参数 target 传进去，然后操作实例化出来的 Thread 对象就能满足要求。但对于更复杂的需求，还是继承于 Thread 类再自己实现一些功能更方便。

依赖注入

依赖注入是一种设计模式，将对象的依赖关系从内部转移到外部，由外部代码来管理和传递依赖对象。

正文中提到的 API 类，实例化时需要传入发送请求时需要的 requests.Session 对象，这就是一种依赖注入。

依赖注入有助于提高代码的可维护性、可测试性和灵活性。

一些小花招

_

临时变量，或者需要忽略的变量可以用且仅用单个下划线 _ 命名。

比如……

# 在仅需要 walk *文件*的时候
for root, _, files in os.walk():
    for file in files:
        ...
# 在解包正则表达式匹配结果的时候
a, _, b, _ = re.findall(r"...", s, re.I)[0] # 没有检查结果，请勿模仿

:=

:= ，「海象运算符」，在 Python 3.8+ 版本中可用，允许在表达式中进行赋值操作，用于减少冗余代码和重复计算次数。

def find_bvid(s: str) -> str | None:
    if m := re.search(r"(?<![A-Za-z])(BV[a-zA-Z0-9]{10})", s):
        return m.group(1)
    return None

算是不错的语法糖，但是同时也需要注意可读性（）

三目运算符

也被称作条件表达式

语法是：

value_if_true if condition else value_if_false

表达式将会根据 condition 的值决定整个表达式的结果是 value_if_true 还是 value_if_false 的值。

一个简单的例子，求得两个值中的较大值：

x = 10
y = 20
max_value = x if x > y else y

可以嵌套使用，但这时可读性很差。这时建议拆成普通的 if-else 系列语句。

别看 有史

# (省略的定义用`...`表示)
...
class VideoWorker(threading.Thread, ...):
    ...
    def _worker(self):
        ...
        finalfile = os.path.join(
            self._savedir,
            filename_escape(
                (
                    f"{title}"
                    + (
                        (f"_P{pindex+1}" if len(cidlist) > 1 else "")
                        if self._correct_pindex is None
                        else f"_P{self._correct_pindex}"
                    )
                    + (f"_{ptitle}" if ptitle != title or self._correct_ptitle else "")
                    + (
                        f"_{bilicodes.stream_dash_audio_quality.get(aqid)}"
                        if self._audio_only
                        else f"_{bilicodes.stream_dash_video_quality.get(vqid)}"
                    )
                )
                + (
                    (".flac" if is_lossless else ".mp3")
                    if self._audio_only
                    else (".mkv" if is_lossless else ".mp4")
                )
            ),
        )
        ...
    ...

懒得喷

tqdm 库

tqdm 库是一个智能的进度条库，主要用于在终端上显示进度条。

最基础的使用方法是直接在 for 循环的 iterable 对象外面套一个 tqdm() （将被迭代对象作为 tqdm 对象实例化的第一个参数），循环的进度就能随着对象的迭代被打印出来：

import time

from tqdm import tqdm  # 惯例写法

text = ""
for char in tqdm(["a", "b", "c", "d"]):
    time.sleep(0.25)
    text = text + char

当然也可以在实例化时传入更多参数，比如描述文本 ( desc )、总进度 ( total )、结束后是否将进度条保留在屏幕上 ( leave )、进度的单位 ( unit )、是否禁用 ( disable )、存在多个进度条时在屏幕上的位置 ( position ) 等等。

当然也可以手动更新进度，这样的做法更常用于下载。这时又可以使用 with 语句：

import requests
from tqdm import tqdm

def download_file(url, filename):
    # 一个最简的带进度条的分块下载实现
    response = requests.head(url)
    file_size = int(response.headers.get('content-length', 0))

    with requests.get(url, stream=True) as r, open(filename, 'wb') as file:
        with tqdm(total=file_size, unit='B', unit_scale=True, desc=filename) as pbar:
            for chunk in r.iter_content(chunk_size=1024):
                if chunk:  # 空块用于保持连接
                    file.write(chunk)
                    pbar.update(len(chunk))

更底层一些，通过手动为 tqdm 实例的 n total desc 等属性设置值，可以任意动态改变进度条的状态；此外使用 refresh() 方法也可以强制刷新进度条的显示。

此外 tqdm 还能作为 CLI 应用程序被直接命令行调用，接收管道数据并显示进度条；还能使用 matplotlib 图形化地展示进度；还能用于异步任务和多进程任务；还与 pandas jupyter dask 等第三方库有非常良好的集成 —— 参见官方文档！

高阶类型标注

TBD...

---

1. 协程是异步的一种实现方式 ↩︎

2. 不过这种情况高性能异步应该也够了大概，在有现成的第三方库支持的情况下 ↩︎

3. 因为每个进程都会获得一个 GIL 而线程不行。当然也可以使用 C/C++ 编写扩展模块来绕过这个限制，或者使用一些高性能第三方库 ↩︎